/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 * Thread management for memcached.
 */
#include <assert.h>
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

#include "../utility/util.h"
#include "../net/event_msgqueue.h"
#include "thread.h"

#include <glog/logging.h>



typedef struct
{
    pthread_t thread_id;        /* unique ID of this thread */
    struct event_base *base;    /* event_base of thread */
    struct event_msgqueue* queue; /* circle queue*/
} LIBEVENT_THREAD;

/*
 * Each libevent instance has a wakeup pipe, which other threads
 * can use to signal that they've put a new connection on its queue.
 */
static LIBEVENT_THREAD *threads;

/*
 * Number of worker threads that have finished setting themselves up.
 */
static int init_count = 0;
static pthread_mutex_t init_lock;
static pthread_cond_t init_cond;
static int num_threads;

extern pthread_t *recv_tids;
extern int recv_num_threads;
extern int recv_quit_flag;

/*
 * Creates a worker thread.
 */
static void create_worker(LIBEVENT_THREAD* thread, void *(*func)(void *), void *arg) 
{
    pthread_attr_t  attr;
    int             ret;

    pthread_attr_init(&attr);

    if ((ret = pthread_create(&thread->thread_id, &attr, func, arg)) != 0) {
        LOG(WARNING) << "Can't create thread: " << strerror(ret);
        exit(EXIT_FAILURE);
    }
}

/*
 * Set up a thread's information.
 */
static void setup_thread(LIBEVENT_THREAD *me, int queue_hwm, int queue_lwm, msgqueue_callback callback, void* cbarg) 
{
    me->base = event_init();
    if (! me->base) {
        LOG(WARNING) << "Can't allocate event base";
        exit(EXIT_FAILURE);
    }
    me->queue = msgqueue_new(me->base, queue_hwm, queue_lwm, false, callback, cbarg);
    if (me->queue == NULL) {
        LOG(WARNING) << "Failed to allocate memory for queue: " << strerror(errno);
        exit(EXIT_FAILURE);
    }
}


/*
 * Worker thread: main event loop
 */
static void *worker_libevent(void *arg) 
{
    LIBEVENT_THREAD *me = (LIBEVENT_THREAD*)arg;
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    /* Any per-thread setup can happen here; thread_init() will block until
     * all threads have finished initializing.
     */

    pthread_mutex_lock(&init_lock);
    init_count++;
    pthread_cond_signal(&init_cond);
    pthread_mutex_unlock(&init_lock);

    event_base_loop(me->base, 0);
    return NULL;
}

/* Which thread we assigned for a request (connection)*/

int thread_dispatch_one(int idx, void* data)
{
    LIBEVENT_THREAD* thread = threads + idx;
    return msgqueue_push(thread->queue, data);
}

void thread_destroy()
{
    recv_quit_flag = 1; 
    for (int i = 0; i < recv_num_threads; i++) {
        pthread_join(recv_tids[i], NULL);
    }


    for (int i = 0; i < num_threads; i++) {
        LIBEVENT_THREAD *thread = threads + i;

        if (thread->queue == NULL){
            continue;
		}

        msgqueue_destroy(thread->queue);
    }
}

/*
 * Initializes the thread subsystem, creating various worker threads.
 *
 * nthreads  Number of worker event handler threads to spawn
 */
void thread_init(int nthreads, int queue_hwm, int queue_lwm, msgqueue_callback callback, void* cbarg)
{
    int i;

    pthread_mutex_init(&init_lock, NULL);
    pthread_cond_init(&init_cond, NULL);
    num_threads = nthreads;

    threads = (LIBEVENT_THREAD*)calloc(nthreads, sizeof(LIBEVENT_THREAD));
    if (! threads) {
        perror("Can't allocate thread descriptors");
        exit(1);
    }

    for (i = 0; i < nthreads; i++) {
        setup_thread(&threads[i], queue_hwm, queue_lwm, callback, cbarg);
    }

    /* Create threads after we've done all the libevent setup. */
    for (i = 0; i < nthreads; i++) {
        create_worker(&threads[i], worker_libevent, &threads[i]);
    }

    /* Wait for all the threads to set themselves up before returning. */
    pthread_mutex_lock(&init_lock);
    while (init_count < nthreads) {
        pthread_cond_wait(&init_cond, &init_lock);
    }
    pthread_mutex_unlock(&init_lock);
}
